-module(map_server).
-include_lib("kernel/include/file.hrl").
%-behaviour(gen_server).
%-export([start_link/0]).

%start_link() ->
%    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%init(_Args) ->
%    {ok, []}.
-export([map_func/5]).

mapper(_Function, []) ->
    [];
mapper(Function, [{Key, Value}|T]) ->
    L = Function({Key, Value}),
    if
	L /= [] ->
	    gen_server:call(partition_buffer, {newinput, L})
    end,
    mapper(Function, T).

map_func(File_path, Start, End, Mapper, ReducerList) ->
    {ok, FileInfo} = file:read_file_info(File_path),
    {ok, InputDevice} = file:open(File_path, [read, binary, raw]),
    ReadLen = End - Start,
    FileLength = FileInfo#file_info.size,
    read_local(0, InputDevice, ReadLen, FileLength, ReducerList, Mapper),
    file:close(InputDevice).

read_local(Remain, InputDevice, ReadLen, FileLen, ReducerList, Mapper) ->
    if
	Remain > FileLen ->
	    gen_server:call(partition_buffer, {over});
	true ->
	    {ok, BinaryData} = file:pread(InputDevice, Remain, ReadLen),
	    InputData = string:tokens(binary_to_list(BinaryData),"\r\n"),
	    KVInputData = inputformat:text_input_format(InputData),
	    partition_buffer:start_link(ReducerList),
	    mapper(Mapper, KVInputData),
	    read_local(Remain+ReadLen, InputDevice, ReadLen ,FileLen, ReducerList, Mapper)
    end.

